package com.zhengbo.flink.demo.mq.cardata;

import com.alibaba.fastjson.JSONObject;
import com.zhengbo.flink.demo.base.mongo.MongoService;
import com.zhengbo.flink.demo.base.utils.SpringContextUtil;
import com.zhengbo.flink.demo.mq.bean.CarStatus;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.IterationRuntimeContext;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

/**
 * 车辆数据用于存储 类似于jstorm中的bolt
 *
 * @author zhengbo
 * @date 2019/8/15
 */
@Slf4j
public class CarDataSinkFunction extends RichSinkFunction<CarStatus> {

    private MongoService mongoService;

    @Override
    public void open(Configuration parameters) throws Exception {

        super.open(parameters);

        log.info("CarDataSinkFunction open....");


        mongoService = SpringContextUtil.getBean(MongoService.class);
        log.info("CarDataSinkFunction open success....");
    }

    @Override
    public void close() throws Exception {
        super.close();
    }

    public CarDataSinkFunction() {
        super();
    }

    @Override
    public void setRuntimeContext(RuntimeContext t) {
        super.setRuntimeContext(t);
    }

    @Override
    public RuntimeContext getRuntimeContext() {
        return super.getRuntimeContext();
    }

    @Override
    public IterationRuntimeContext getIterationRuntimeContext() {
        return super.getIterationRuntimeContext();
    }

    @Override
    public void invoke(CarStatus value, Context context) throws Exception {

        log.info("invoke carStatus:{},context:{}", value, context);

        mongoService.insert(JSONObject.toJSONString(value), "flink-test-carData");

    }
}
